package com.oraclefmw.crusher.core;

import java.util.Hashtable;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;


public class JMSWorker extends AbstractWorker{

	public static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
	public static String JMS_FACTORY = "jms.queue.FileTransferConnectionFactory";
	public static String QUEUE = "jms.queue.FileTransferQueue";
	public static String SERVER_URL = "t3://localhost:7001";

	private QueueConnectionFactory qconFactory;
	private QueueConnection qcon;
	private QueueSession qsession;
	private QueueSender qsender;
	private Queue que;
	private BytesMessage message;

	private static Context context;

	public JMSWorker(BlockingQueue<Fragment> queue, Properties props) {
		
		super( queue, props );		
	}

	public void check() {
		
		super.check();
		
		if (props.getProperty("jndi_factory") != null)
			JNDI_FACTORY = (String) props.getProperty("jndi_factory");
		if (props.getProperty("jms_factory") != null)
			JMS_FACTORY = (String) props.getProperty("jms_factory");
		if (props.getProperty("jms_queue") != null)
			QUEUE = (String) props.getProperty("jms_queue");
		if (props.getProperty("server_url") != null)
			SERVER_URL = (String) props.getProperty("server_url");
	}
	
	public void init() {

		try {
			
			if (context == null) {

				Hashtable<String, String> env = new Hashtable<String, String>();
				env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
				env.put(Context.PROVIDER_URL, SERVER_URL);
				context = new InitialContext(env);

			}

			qconFactory = (QueueConnectionFactory) context.lookup(JMS_FACTORY);
			qcon = qconFactory.createQueueConnection();
			qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			que = (Queue) context.lookup(QUEUE);
			qsender = qsession.createSender(que);
			message = qsession.createBytesMessage();
			qcon.start();

		} catch (Exception e) {

			System.err.println("jms worker initialize error:" + e.getMessage());
			e.printStackTrace();
		}
	}

	public void send(Fragment fragment) {

		try {
			message.setLongProperty("Begin", fragment.getBegin());
			message.setLongProperty("Length", fragment.getContent().length);
			message.setStringProperty("FileName", fragment.getFileName());

			message.writeBytes(fragment.getContent());
			qsender.send(message);

			message.clearBody();
			message.clearProperties();

		} catch (JMSException e) {

			System.err.println("sending jms message error:" + e.getMessage());
			e.printStackTrace();
		}
	}

	public void close() {

		try {

			qsender.close();
			qsession.close();
			qcon.close();
		} catch (JMSException e) {

			System.err.println("closing jms connection error:" + e.getMessage());
			e.printStackTrace();
		}
	}

	@Override
	public void run() {

		check();
		
		init();

		for (int i = 0; i < pageSize; i++) {

			Fragment fragment = null;
			
			try {
				fragment = queue.take();
			} catch (InterruptedException e) {
				//
			}

			if (fragment != null) {

				send(fragment);
			}
		}
		close();
	}
}
